package com.zte.claa.inficombo.csa.app.service.sapi.ffhost;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.script.Invocable;

import org.apache.mina.core.future.WriteFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.config.FFTransDevConnProps;
import com.zte.claa.inficombo.csa.app.dao.mysql.impl.FFResDevMapper;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.jsplugin.CSAJSPluginService;
import com.zte.claa.inficombo.csa.app.service.jsplugin.JSParam;
import com.zte.claa.inficombo.csa.app.service.sapi.ffhost.bean.FFResDevBase;
import com.zte.claa.inficombo.csa.app.service.sapi.ffhost.mina.FFTransDevConnServerManager;
import com.zte.claa.inficombo.csa.app.service.sapi.ffhost.mina.FFTransDevMessage;

@Service
public class FFTransDevSAPIServiceImpl implements FFTransDevSAPIService, InitializingBean, DisposableBean {
	// js插件中上报数据解析的function名称，统一固定命名 。
	private static final String UPDATA_PARSE_FUNC = "funcupdataparse";
	// 下行命令样例： downcmd61，61=数据单元类型标记.
	private static final String DOWN_CMD_FUNC_PREFIX = "downcmd";
		
	// 日志.
	private Logger logger = LoggerFactory.getLogger(this.getClass());
	
	// 连接参数.
	@Autowired
	private FFTransDevConnProps connProps;
	
	@Autowired
	private CSAJSPluginService jsPnpService;
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private FFResDevMapper ffResMapper;
	
	@Autowired
	private FFDataPeristentService persistentSvr;
	
	// 连接管理服务器
	private FFTransDevConnServerManager connSvrManager = null;
	
	// 上行数据处理线程池.
	private ExecutorService dataRptExecSvr;
	
	// conn-sessionid <--> transdeveui的映射表. key: sessionId, val: transdevEui.
	private Map<Long, String> sessiondevMap = null;
	
	@Override
	public void afterPropertiesSet() throws Exception {
		if(this.connProps.isSwitchon() == false){
			logger.warn("FFTransDevSAPIService is set OFF.");
			return;
		}
		try{
			this.sessiondevMap = new ConcurrentHashMap<Long, String>();
			// 上行消息分发线程池.
			RejectedExecutionHandler faultHandler = new ThreadPoolExecutor.DiscardPolicy();
			this.dataRptExecSvr = new ThreadPoolExecutor(1, 
				this.connProps.getUpdealermaxnum(), 120, TimeUnit.MILLISECONDS, 
    			new LinkedBlockingQueue<Runnable>(this.connProps.getMsgqueuesize()), faultHandler);
			// 初始化connserver.
			this.connSvrManager = new FFTransDevConnServerManager(this.connProps, this);
			this.connSvrManager.start();
			logger.info("FFTransDevSAPIService init ok!");
			// just for test
			// test();
		}catch(Exception e){
			logger.error("FFTransDevSAPIService init failed!", e);
			this.destroy();
		}
	}
	
	private void test(){
		Thread t = new Thread(new Runnable(){
			@Override
			public void run() {
				sendDemoData();
			}
		});
		t.start();
	}
	
	private void sendDemoData(){
		try{
			Thread.sleep(3000);
		}catch(Exception e){}
		long sessionId = 1;
		FFTransDevMessage message = new FFTransDevMessage();
		message.setMsgTime(System.currentTimeMillis());
		message.setCmd((byte)FFTransDevMessage.CMD_RPTDATA);
		message.setSno(1);
		message.setSrcAddrBytes(new byte[]{1,0,0,0,0,0});
		message.setDstAddrBytes(new byte[]{1,0,0,0,0,0});
		byte[] data = new byte[12];
		// typeflag
		data[0] = 1;
		// msgnum
		data[1] = 1;
		// systype
		data[2] = 1;
		// sysaddr
		data[3] = 1;
		// sys-state [2]
		data[4] = 3;
		data[5] = 2;
		// datetime
		FFDateTime ffdt = new FFDateTime();
		ffdt.setMsgTime(System.currentTimeMillis());
		data[6] = ffdt.getSecond();
		data[7] = ffdt.getMinute();
		data[8] = ffdt.getHour();
		data[9] = ffdt.getDay();
		data[10] = ffdt.getMonth();
		data[11] = ffdt.getYear();
		// 设置数据》
		message.setData(data);
		// 设置校验和.
		message.setCheckSum(message.calculateSumCheck());
		this.onFFTransDevMessageReport(sessionId, message);
		logger.info("Send demo data ok!");
	}
	               
	@Override
	public void destroy() throws Exception {
		try{
			if(this.connSvrManager != null) {
				this.connSvrManager.stop();
				this.connSvrManager = null;
			}
			if(this.dataRptExecSvr != null){
				this.dataRptExecSvr.shutdown();
				this.dataRptExecSvr = null;
			}
			if(this.sessiondevMap != null){
				this.sessiondevMap.clear();
				this.sessiondevMap = null;
			}
		}catch(Exception e){
			logger.error("destroy failed!", e);
		}
	}

	/**
	 * 接收到用户信息传输装置上报的消息
	 * 
	 * 主要逻辑：
	 * 1）消息进队列；
	 * 2）根据sessionId和message中携带的源地址，找到对应的传输装置实例，并绑定；如果已经绑定，则不需要重复处理；
	 * 3）线程池解析消息；
	 * 4）对于上报消息，需要给出确认响应；
	 * 5）非法消息的处理（打印日志？给出否认应答）；
	 * 
	 */
	@Override
	public void onFFTransDevMessageReport(long sessionId, FFTransDevMessage message) {
		if(this.dataRptExecSvr != null){
			FFTransDevRptMsg rptmsg = new FFTransDevRptMsg();
			rptmsg.setMessage(message);
			rptmsg.setSessionId(sessionId);
			this.dataRptExecSvr.execute(new Runnable(){
				@Override
				public void run() {
					dispatchRptMessage(rptmsg);
				}
			});
		} else {
			logger.warn("Rpt message dispatcher pool is null!");
		}
	}
	
	/**
	 * 上报消息分发处理.
	 * 0）更新session -->transdevEui的关系
	 * 1）对于上报消息（命令类型=2），需要给予确认消息； 
	 * 2）调用js插件解析上报消息：区分上报消息2；查询应答5/否认6；命令确认3/否认6；其他消息；
	 * 3）持久化上报消息（日志数据），告警数据需要对外发布；
	 * @param rptmsg
	 */
	private void dispatchRptMessage(FFTransDevRptMsg rptmsg){
		FFTransDevMessage orgdata = rptmsg.getMessage();
		long sessionId = rptmsg.getSessionId();
		logger.info("Begin process rpt msg: sessionid={}, msgno={}.", sessionId, orgdata.getSno());
		// 1. 尝试更新session -->transdevEui的关系
		String transdevEui = this.sessiondevMap.get(rptmsg.getSessionId());
		if(transdevEui == null){
			// 加载transdev数据
			transdevEui = this.loadTransdevData(sessionId, orgdata.getSrcAddr());
			logger.info("load transdev data ok: transdeveui={}", transdevEui);
			// 更新状态connstate...
			if(transdevEui != null && !transdevEui.equals("")){
				try {
					this.ffResMapper.updateFFTransDevState(transdevEui, FFResDevBase.CONN_STATE_NORMAL);
				} catch (Exception e) {
					logger.error("1.updateFFTransDevState (normal) failed: transdeveui={}", transdevEui, e);
				}
			} else {
				logger.warn("FFTransDev not exists and discard rptmsg: "+orgdata.toString());
				return;
			}
		}
		// 2. 检查消息的合法性，并对上报消息予以确认.
		int flag = this.checkAndAckRptMessage(sessionId, orgdata);
		logger.info("checkAndAckRptMessage flag={}", flag);
		// 3. 解析数据并持久化.
		this.parseRptData(transdevEui, orgdata);
	}
	
	// 加载transdev数据，更新sessionid -->transdeveui的映射关系.
	private String loadTransdevData(long sessionId, String transDevAddr){
		FFResDevBase transDev = this.cacheManager.getFFTransDevByAddr(transDevAddr);
		if(transDev == null){
			logger.warn("FFTransdev resource dev not exist for addr: "+transDevAddr);
			return null;
		}
		this.sessiondevMap.put(sessionId, transDev.getDeveui());
		return transDev.getDeveui();
	}
	
	// 检查消息的合法性，并对上报消息予以确认.
	private int checkAndAckRptMessage(long sessionId, FFTransDevMessage orgdata) {
		int ret = 0;
		if(orgdata.startFlagCheck() == false){
			logger.warn("Start flag is invalid: "+orgdata.getStartB1()+""+orgdata.getStartB2());
			ret = 1;
		}
		if(orgdata.endFlagCheck() == false){
			logger.warn("End flag is invalid: "+orgdata.getEndB1()+""+orgdata.getEndB2());
			ret = 2;
		}
		if(orgdata.sumCheckFlagCheck() == false){
			logger.warn("Sumcheck is invalid: rpt["+orgdata.getCheckSum()+"] != calculate["+orgdata.calculateSumCheck()+"]");
			ret = 3;
		}
		if(orgdata.getCmd() == FFTransDevMessage.CMD_RPTDATA){
			logger.info("Try to send ack for report data: sessionid={}, sno={}", sessionId, orgdata.getSno());
			FFTransDevMessage ackMsg = new FFTransDevMessage();
			ackMsg.setData(new byte[0]);
			// 对调地址域.
			ackMsg.setSrcAddr(orgdata.getDstAdrr());
			ackMsg.setDstAdrr(orgdata.getSrcAddr());
			// 复制流水号.
			ackMsg.setSno(orgdata.getSno());
			ackMsg.setMainVersion(orgdata.getMainVersion());
			ackMsg.setUserVersion(orgdata.getUserVersion());
			// 设置时间戳.
			ackMsg.setMsgTime(System.currentTimeMillis());
			// 响应命令.
			short cmd = (ret == 0) ? FFTransDevMessage.CMD_CONFIRM : FFTransDevMessage.CMD_DENYACK; 
			ackMsg.setCmd((byte)cmd);
			// 最后计算校验和.
			ackMsg.setCheckSum(ackMsg.calculateSumCheck());
			try {
				this.connSvrManager.sendCommand(sessionId, ackMsg);
				logger.info("Send ack message ok: "+ackMsg);
			} catch (Exception e) {
				logger.error("Send ack message failed: "+ackMsg, e);
			}
		}
		return 0;
	}
	
	// 调用插件解析上报数据.
	private void parseRptData(String transdevEui, FFTransDevMessage orgdata) {
		try{
			String devTypeId = this.connProps.getTransdevtype().trim();
			Invocable jsRunner = this.jsPnpService.getJsPLugin(devTypeId);
			if(jsRunner == null){
				logger.error("[*!!*]Can not find the JS plugin for devtype="+devTypeId);
				return;
			}			
			byte[] payloadBS = orgdata.getData();
			JSParam jsParam = new JSParam();
			jsParam.appendPara(JSParam.UPDATA_PARA_NAME, payloadBS);
			Object rptData = jsRunner.invokeFunction(UPDATA_PARSE_FUNC, jsParam);
			// 情况1--解析结果无效；
			if(rptData == null){
				logger.info("[*!*]Invalid or nouse report data!");
				return;
			}
			FFTransDevRptParseResult parseRet = (FFTransDevRptParseResult)rptData;
			Map<String, Object> datamap = parseRet.getDatamap();
			if(datamap == null || datamap.size() == 0){
				logger.info("There is no data in parse resultmap.");
				return;
			}
			Object obj = datamap.get(FFTransDevRptParseResult.RET_KEY_METEDATA);
			if(obj != null){
				this.processRptMetedata(transdevEui, parseRet.getDataflag(), obj);				
			}
			obj = datamap.get(FFTransDevRptParseResult.RET_KEY_METEDATALIST);
			if(obj != null){
				this.processRptMetedataList(transdevEui, parseRet.getDataflag(), obj);				
			}
		}catch(Exception e){
			logger.error("parseRptData failed: transdeveui={}, data={}", transdevEui, orgdata.toString(), e);
		}
	}
	
	// 处理解析到的log数据
	private void processRptMetedata(String transdevEui, int dataflag, Object metedataObj) {
		if(metedataObj == null || metedataObj.equals("")){
			logger.info("parse result data is null!");
			return;
		}
		try{
			this.persistentSvr.saveFFResDevRptData(transdevEui, dataflag, metedataObj);
		}catch(Exception e){
			logger.error("processRptMetedata failed: dataflag="+dataflag+", data="+metedataObj, e);
		}
	}
	
	// 处理解析到的log数据
	private void processRptMetedataList(String transdevEui, int dataflag, Object metedataListObj) {
		if(metedataListObj == null){
			logger.info("parse result data is null!");
			return;
		}
		try{
			this.persistentSvr.saveFFResDevRptDataByBatch(transdevEui, dataflag, metedataListObj);			
		}catch(Exception e){
			logger.error("processRptMetedata failed: dataflag="+dataflag+", data="+metedataListObj, e);
		}
	}

	// 底层上报连接会话关闭.
	@Override
	public void onConnectionClose(long sessionId) {
		String transDevEui = this.sessiondevMap.remove(sessionId);
		if(transDevEui != null){
			logger.warn("Connection is closed: transdeveui={}", transDevEui);
			try {
				this.ffResMapper.updateFFTransDevState(transDevEui, FFResDevBase.CONN_STATE_BROKE);
			} catch (Exception e) {
				logger.error("2.updateFFTransDevState (broken) failed: transdeveui={}", transDevEui, e);
			}
		}		
	}

	@Override
	public void onExceptionReport(long sessionId, Throwable cause) {
		logger.error("Exception is reported: sessionId={}", sessionId, cause);
	}

	// 上层应用主动关闭链路.
	@Override
	public void closeConnection(String transDevEui) throws Exception {
		// transDevEui-->sessionId-->关闭.
		if(this.sessiondevMap == null){
			return;
		}
		// 设置transdev的链路断开...
		try {
			this.ffResMapper.updateFFTransDevState(transDevEui, FFResDevBase.CONN_STATE_BROKE);
		} catch (Exception e) {
			logger.error("3.updateFFTransDevState (broken) failed: transdeveui={}", transDevEui, e);
		}
		Iterator<Long> it = this.sessiondevMap.keySet().iterator();
		while(it.hasNext()){
			long sessionId = it.next();
			String devEui = this.sessiondevMap.get(sessionId);
			if(devEui == null || !transDevEui.equals(devEui)){
				continue;
			}
			this.connSvrManager.closeConnServer(sessionId);
			// 删除对应的映射.
			it.remove();
		}
	}
	
	// 根据 transdevEui 查找对应的sessionId
	private long getSessionIdByTransdev(String transdevEui) {
		if(this.sessiondevMap == null){
			return -1;
		}
		Iterator<Long> it = this.sessiondevMap.keySet().iterator();
		while(it.hasNext()){
			long sessionId = it.next();
			String devEui = this.sessiondevMap.get(sessionId);
			if(devEui != null && transdevEui.equals(devEui)){
				return sessionId;
			}
		}
		return -1;
	}

	@Override
	public WriteFuture sendCommand(String transDevEui, FFTransDevCommand cmd) throws Exception {
		if(transDevEui == null || transDevEui.equals("")){
			logger.warn("sendCommand failed: transDevEui is null");
			throw new Exception("sendCommand failed: transDevEui is null!");
		}
		if(cmd == null){
			logger.warn("sendCommand failed: cmd is null");
			throw new Exception("sendCommand failed: cmd is null!");
		}
		FFResDevBase transDev = this.cacheManager.getFFTransDevByEui(transDevEui);
		if(transDev == null){
			logger.warn("FFTransDev not exist: "+transDevEui);
			throw new Exception("FFTransDev not exist: "+transDevEui);
		}
		long sessionId = this.getSessionIdByTransdev(transDevEui);
		if(sessionId == -1){
			logger.warn("Session for FFTransDev not exist: "+transDevEui);
			throw new Exception("传输装置尚未建立连接: "+transDevEui);
		}
		String devTypeId = transDev.getDevtype();
		Invocable jsRunner = this.jsPnpService.getJsPLugin(devTypeId);
		if(jsRunner == null){
			logger.error("[*!!*]Can not find the JS plugin for devtype="+devTypeId);
			throw new Exception("JS Plugin not exist: "+devTypeId);
		}
		String cmdFuncName = DOWN_CMD_FUNC_PREFIX+cmd.getCmdcode();
		JSParam jsParam = new JSParam();
		Map<String, Object> paraMap = new HashMap<String, Object>();
		paraMap.put("cmdcode", cmd.getCmdcode());
		paraMap.put("typeflag", cmd.getTypeflag());
		if(cmd.getParas() != null){
			paraMap.putAll(cmd.getParas());
		}
		jsParam.setParas(paraMap);
		byte[] cmdpayload = (byte[])jsRunner.invokeFunction(cmdFuncName, jsParam);
		if(cmdpayload == null || cmdpayload.length == 0){
			logger.error("[*!!*]Cmd build in JSPlugin failed: "+cmdFuncName);
			throw new Exception("Cmd build in JSPlugin failed: "+cmdFuncName);
		}
		FFTransDevMessage msg = new FFTransDevMessage();
		msg.setCmd((byte)cmd.getCmdcode());
		msg.setMsgTime(System.currentTimeMillis());
		// 这里不设置命令的sno，connsvr会判断并设置.
		msg.setSrcAddr(this.connProps.getLocaladdr().trim());
		msg.setDstAdrr(transDev.getTransdevaddr());
		// 查询传输装置的系统状态.
		msg.setData(cmdpayload);
		// 消息发送之前connsvr自动计算和设置校验和.
		// msg.setCheckSum(msg.calculateSumCheck());
		return this.connSvrManager.sendCommand(sessionId, msg);
	}

}
